import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;

public class Hw1Grp5 {
    //  for HBase start and some operations
    public static String tableName = "Result";
    public static String columnFamily = "res";
    public static HTableDescriptor htd;
    public static HColumnDescriptor cf;
    public static Configuration configuration;
    public static HBaseAdmin hAdmin;
    public static HTable table;
    //    parameters
    public static String inputFile;
    public static ArrayList<Integer> distinctColumnIndex = new ArrayList<>();
    public static ArrayList<String> distinctColumnName = new ArrayList<>();
    public static int selectColumnIndex;
    public static double cmpNum;
    public static String operation;

    public static HashSet<ArrayList<String>> distinctRow = new HashSet<>();
    public static double min = 0.0000001;



    public static void initInsert() throws IOException {
        htd = new HTableDescriptor(TableName.valueOf(tableName));
        cf = new HColumnDescriptor(columnFamily);
        htd.addFamily(cf);
        configuration = HBaseConfiguration.create();
        hAdmin = new HBaseAdmin(configuration);
        if (hAdmin.tableExists(tableName)) {
            System.out.println("Table exists!");
            return;
        }

        hAdmin.createTable(htd);
        hAdmin.close();
        table = new HTable(configuration, tableName);
        System.out.println("Initialization finished.");
    }

    public static void endInsert() throws IOException {
        table.close();
        System.out.println("Operation finished.");
    }
    //    parse command
//     R=/hw1/distinct_1.tbl select:R1,ge,30 distinct:R2,R0
    public static void parseCommand(String[] args) {
        for (String str : args) {
            if (str.contains("R=")) {
                inputFile = str.substring(2);
            } else if (str.contains("select:")) {
                selectColumnIndex = Integer.parseInt(str.substring(8,9));
                operation = str.substring(10,12);
                cmpNum = Double.parseDouble(str.substring(13));
            } else if (str.contains("distinct:")) {
                int r = str.indexOf("R");
                while (r != -1) {
                    int i = locateI(str, r+1);
                    int index = Integer.parseInt(str.substring(r+1, i));
                    distinctColumnIndex.add(index);
                    r = str.indexOf("R", r+1);
                }
            } else {
                System.out.println("Command incomplete!");
            }
        }
        for (int i : distinctColumnIndex) {
            distinctColumnName.add("R" + i);
        }
    }

    public static boolean satisfyCondition(double selectKey) {
        boolean ret;
        switch (operation) {
            case "ge":
                ret = selectKey >= cmpNum;
                break;
            case "gt":
                ret = selectKey > cmpNum;
                break;
            case "eq":
                ret = Math.abs(selectKey - cmpNum) < min;
                break;
            case "lt":
                ret = selectKey < cmpNum;
                break;
            case "le":
                ret = selectKey <= cmpNum;
                break;
            case "ne":
                ret = selectKey != cmpNum;
                break;
            default:
                throw new IllegalStateException("Unexpected value: " + operation);
        }
        return ret;
    }
    //    maek a hashmap
    public static void formHashSet() throws IOException, URISyntaxException {
        BufferedReader in = readFile(inputFile);
        String str;
        while ((str = in.readLine()) != null) {
            String[] cols = str.split("\\|");
            double selectKey = Double.parseDouble(cols[selectColumnIndex]);
            if (satisfyCondition(selectKey)) {
                ArrayList<String> row = new ArrayList<>();
                for (int i : distinctColumnIndex) {
                    row.add(cols[i]);
                }
                distinctRow.add(row);
            }
        }
    }

    public static void insertRow() throws InterruptedIOException, RetriesExhaustedWithDetailsException {
        int i = 0;
        for (ArrayList<String> strs : distinctRow) {
            for (int j=0; j<distinctColumnName.size(); ++j) {
                String columnName = distinctColumnName.get(j);
                String content = strs.get(j);
                Put p = new Put(Bytes.toBytes(i));
                p.add(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(content));
                table.put(p);
            }
            i++;
        }
    }

    //read from local
    public static BufferedReader readFileFromLocal(String path) throws IOException {
        return new BufferedReader(new FileReader(path));
    }
    //read from hdfs
    public static BufferedReader readFile(String path) throws IOException, URISyntaxException {
        String file = "hdfs://localhost:9000";
        file += path;
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(file), conf);
        FSDataInputStream in_stream = fs.open(new Path(file));
        return new BufferedReader(new InputStreamReader(in_stream));
    }
    //for parse command
    public static int locateI(String str, int start) {
        int i;
        for (i=start; i<str.length(); ++i) {
            if (str.charAt(i) < '0' || str.charAt(i) > '9') {
                break;
            }
        }
        return i;
    }

    public static void main(String[] args) throws IOException, URISyntaxException {
        parseCommand(args);
        initInsert();
        formHashSet();
        insertRow();
        endInsert();



    }
}